package com.lingjtx.common.thread.pooling;

import com.lingjtx.common.thread.core.PoolNameConst;
import com.lingjtx.common.thread.core.ThreadPoolManager;
import com.lingjtx.common.thread.mq.RabbitStreamConsumerProperties;
import com.rabbitmq.stream.Environment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RabbitStreamEnvironmentFactory implements ObjectFactory<Environment> {
    private static final Logger log = LoggerFactory.getLogger(RabbitStreamEnvironmentFactory.class);

    private final ThreadPoolManager threadPoolManager;
    private final RabbitStreamConsumerProperties props;

    public RabbitStreamEnvironmentFactory(ThreadPoolManager threadPoolManager, RabbitStreamConsumerProperties props) {
        this.threadPoolManager = threadPoolManager;
        this.props = props;
    }

    @Override
    public Environment create() {
        return Environment.builder()
                .host(props.getHost())
                .port(props.getPort())
                .username(props.getUsername())
                .password(props.getPassword())
                .scheduledExecutorService(threadPoolManager.getOrCreateScheduled(PoolNameConst.RABBIT_STREAM_SCHEDULER))
                .build();
    }

    @Override
    public boolean validate(Environment env) {
        return env != null;
    }

    @Override
    public void destroy(Environment env) {
        shutdown();
        if (env != null) {
            try {
                env.close(); // Environment 实现了 AutoCloseable
            } catch (Exception e) {
                // 打印日志或记录
                log.warn("Failed to close RabbitMQ Stream Environment", e);
            }
        }
    }

    @Override
    public void shutdown() {
        threadPoolManager.shutdown(PoolNameConst.RABBIT_STREAM_SCHEDULER);
    }
}
